package com.dms.consumer;

import com.dms.conf.ConfigLoader;
import com.dms.conf.DmsConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Properties;

public class DmsConsumer {

    private static final Logger logger = LoggerFactory.getLogger(DmsConsumer.class);
    private KafkaConsumer<Object, Object> consumer;

    DmsConsumer(String path) throws IOException {
        Properties props = new Properties();
        InputStream in = null;
        try {
            in = new BufferedInputStream(new FileInputStream(path));
            props.load(in);
            consumer = new KafkaConsumer<Object, Object>(props);
        } catch (IOException e) {
            logger.error("fail to read config file.");
        } finally {
            if (in != null) {
                in.close();
            }
        }

    }

    DmsConsumer() throws Exception {
        Properties props = null;
        try {
            props = ConfigLoader.loadFromClasspath(DmsConfig.CONFIG_CONSUMER_FILE_NAME);
            consumer = new KafkaConsumer<Object, Object>(props);
        } catch (IOException e) {
            logger.error("fail to read config file.{}",e);
        }
    }

    public DmsConsumer(String groupId, String path) throws Exception {
        Properties props = null;
        try {
            props = ConfigLoader.loadFromClasspath(path);
            props.setProperty("group.id", groupId);
            consumer = new KafkaConsumer<Object, Object>(props);
        } catch (IOException e) {
            logger.error("fail to read config file.{}",e);
        }
    }

    public void consume(List topics) {
        consumer.subscribe(topics);
    }

    public ConsumerRecords<Object, Object> poll(long timeout) {
        return consumer.poll(timeout);
    }

    public void close() {
        consumer.close();
    }
}
